package com.ls.rabbitmqService;

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeoutException;

/**
 * rabbitmq tcp连接池的管理
 * 暂时只考虑单机的情况
 */
public class RabbitmqConnectionPool {

    static Logger logger = LoggerFactory.getLogger(RabbitmqConnectionPool.class);
    static ConnectionFactory factory = null;
    //最大tcp连接数
    public static int maxConnectionNum=3;
    //记录当前所有连接数
    static List<RabbitmqConnection> connList=null;
    //当前tcp连接数
    static int currentConnectionNum;
    //进程启动之后条用初始化连接工厂
    public static void initConnectionFactory(String username,String password,String virtualHost,String host,int port){
        factory= new ConnectionFactory();
        connList=new ArrayList<RabbitmqConnection>();
        factory.setUsername(username);
        factory.setPassword(password);
        factory.setVirtualHost(virtualHost);
        factory.setHost(host);
        factory.setPort(port);
    }
    public static final class RabbitmqConnection{
        private Connection connection;
        private int id;
        //当前连接开辟的信道个数
        private volatile int newChannelNum;

        public RabbitmqConnection(Connection connection, int id) {
            this.connection = connection;
            this.id = id;
        }

        public Connection getConnection() {
            return connection;
        }

        public void setConnection(Connection connection) {
            this.connection = connection;
        }

        public int getId() {
            return id;
        }

        public void setId(int id) {
            this.id = id;
        }

        public int getNewChannelNum() {
            return newChannelNum;
        }
        public int incrNewChannelNum(){
                return newChannelNum++;
        }
        public int decrNewChannelNum(){
                return newChannelNum--;
        }
        public void close(){
            if(connection!=null){
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

    }

    public static List<RabbitmqConnection> getConnections(){
        return connList;
    }

    private static void sort(List<RabbitmqConnection> vector){
        if(vector!=null && vector.size()>0){
            Collections.sort(vector, new Comparator<RabbitmqConnection>() {
                @Override
                public int compare(RabbitmqConnection o1, RabbitmqConnection o2) {
                    if(o1!=null && o2!=null){
                        return Integer.compare(o1.getNewChannelNum(),o2.getNewChannelNum());
                    }
                    return 0;
                }
            });
        }
    }
    public static  RabbitmqConnection getConnection(){
        RabbitmqConnection connection=null;
        if(factory==null){
            logger.error("ConnectionFactory is null");
        }else{
            try {
                if(currentConnectionNum<maxConnectionNum){
                    if(currentConnectionNum>=1 && connList.size()>=1){ //已经有一个tcp连接的情况下,如果这个tcp连接的channnel没有达到最大数，没必要在申请tcp
                        sort(connList);
                        connection=connList.get(0);
                        //一个tcp连接上允许的最大channle数量是有限的
                        //logger.info("getNewChannelNum:"+connection.getNewChannelNum());
                        if(connection.getNewChannelNum()<RabbitmqChannelPool.maxTcpChannelNum && connection.getNewChannelNum()<connection.getConnection().getChannelMax()){
                            return connection;
                        }
                    }
                    connection=new RabbitmqConnection(factory.newConnection(),++currentConnectionNum);
                    connList.add(connection);
                }else{
                    sort(connList);
                    //找到某个tcp连接上创建channel对象较少的
                    connection=connList.get(0);
                }
            } catch (IOException e) {
                logger.error(e.getMessage(),e);
            } catch (TimeoutException e) {
                //创建连接超时
                logger.error("create rabbitmq connection timeout,"+e.getMessage(),e);
            }
        }
        return connection;
    }
    //关闭所有连接
    public static void closeAllConnection(){
        if(connList!=null && connList.size()>0){
            for(RabbitmqConnection connection:connList){
                if(connection!=null)
                    connection.close();
            }
            connList.clear();
        }
    }
}
